package com.idroc.cdfs.util;

import java.util.LinkedList;
import java.util.List;
import java.util.SortedMap;
import java.util.TreeMap;


public class ConsistentHashTable {

	/**
	 * 最大支持逻辑节点数量
	 */
	public static final int MAX_LOGICAL_NODE = 65536;
	
	/**
	 * 最大支持物理节点数量
	 */
	public static final int MAX_PHYSICAL_NODE = 65536;
	
	/**
	 * 计算逻辑节点值辅助数组,保存数组下标的2的幂的值
	 */
	private static int[] locateFactor = new int[17];
	 	
	/**
	 * 初始化两个辅助数组
	 */
	static {
		
		for (int i = 0; i < locateFactor.length; i++) {
			int val = (int)Math.pow(2, i);
			locateFactor[i] = val;  
		} 
	}
	
	/**
	 * 逻辑节点与物理节点的映射
	 */
	private int[] logicalNodes = new int[MAX_LOGICAL_NODE];
	
	/**
	 * 物理节点集合
	 */
	private Object[] physicalNodes = new Object[MAX_PHYSICAL_NODE];
	
	/**
	 * 物理节点组成的一致性hash的位置环
	 */
	private TreeMap<Integer, Integer> physicalNodeRing = new TreeMap<Integer, Integer>();
	
	/**
	 * 检查参数是否有效: 是否在一致性hash的环中
	 * @param index 参数
	 */
	public static void checkIndexRange(int index) {

		if (index > 65536 || index < 0) {
			throw new RuntimeException("非法参数index:" + index);
		}
	} 
	
	/**
	 * 获取饱和状态下物理节点对应的逻辑节点
	 * @param physicalNode 物理节点
	 * @return 逻辑节点
	 */
	public static int getLogicNode(int physicalNode) {
		
		checkIndexRange(physicalNode);
		if (physicalNode == 0 || physicalNode == 65536) {
			return 0;
		} 
		
		int logical = -1;
		
		for (int i = 0; i < locateFactor.length; i++) { 
			if (locateFactor[i] > physicalNode) {
				int rangeSize = MAX_LOGICAL_NODE / locateFactor[i];
				int parity = physicalNode % 2; 
				if(parity == 0) { // 偶数
					int remainder = physicalNode - locateFactor[i - 1] + 1;
					logical = 0 + rangeSize * remainder; 
				} else { // 奇数
					int remainder = physicalNode - locateFactor[i - 1];
					logical = MAX_LOGICAL_NODE / 2 + rangeSize * remainder; 
				}
				return logical;
			}
		}		 
		throw new RuntimeException("无法求出物理节点对应的逻辑节点值");  
	} 
	
	/**
	 * ConsistentHashTable增加指定的物理节点
	 * @param no 物理节点
	 * @param node 节点对象
	 */
	public void addPhysicalNode(int no, Object node) {
		
		checkIndexRange(no);
		physicalNodes[no] = node; 
	 	for (int i = 0; i < locateFactor.length; i++) {
	 		if (locateFactor[i] > no) {
	 			for (int j = no; j < MAX_LOGICAL_NODE; j += locateFactor[i]) {
	 				if (logicalNodes[j] < no) {
	 					logicalNodes[j] = no;
	 				}
	 			}
	 		}
	 	}
	 	
	 	// 确定物理节点在一致性hash环中的位置
	 	int logicalNode = getLogicNode(no);
	 	physicalNodeRing.put(Integer.valueOf(logicalNode), Integer.valueOf(no));
	}
	
	/**
	 * ConsistentHashTable移除指定的物理节点
	 * @param no 物理节点
	 * @return 节点对象,如果存在
	 */
	public Object removePhysicalNode(int no) {
		
		checkIndexRange(no);
		for (int i = 0; i < locateFactor.length; i++) {
	 		if (locateFactor[i] > no) {
	 			for (int j = no; j < MAX_LOGICAL_NODE; j += locateFactor[i]) {
	 				if (logicalNodes[j] == no) {
	 					logicalNodes[j] = no - locateFactor[i - 1];
	 				}
	 			}
	 		}
	 	}
		Object rtv = physicalNodes[no];
		physicalNodes[no] = null;
		

	 	// 确定物理节点在一致性hash环中的位置
	 	int logicalNode = getLogicNode(no);
	 	physicalNodeRing.remove(Integer.valueOf(logicalNode));
		return rtv;
	}
	
	/**
	 * 定位逻辑节点在ConsistentHashTable中的物理节点
	 * @param logicalNode 逻辑节点
	 * @return 物理节点
	 */
	public Object locatePhysicalNode(int logicalNode) {
		
		checkIndexRange(logicalNode);
		int index = logicalNodes[logicalNode];
		return physicalNodes[index];
	}
	
	/**
	 * 获取ConsistentHashTable中物理节点关联的逻辑节点集合
	 * @param physicalNode 物理节点
	 * @return 逻辑节点的集合(List)
	 */
	public List<Integer> locateLogicalNode(int physicalNode) {
		
		List<Integer> rtv = new LinkedList<Integer>();
		for (int i = 0; i < MAX_LOGICAL_NODE; i++) {
			if (logicalNodes[i] == physicalNode) {
				rtv.add(Integer.valueOf(i));
			}
		}
		return rtv;
	}
	
	/**
	 * 返回当前逻辑节点数组的镜像
	 * @return 逻辑节点数组镜像数组
	 */
	public int[] getLogicNodeSnapshot() {
		
		int[] newLogicals = new int[logicalNodes.length];
		System.arraycopy(logicalNodes, 0, newLogicals, 0, newLogicals.length);
		return newLogicals;
	} 
	
	/**
	 * 获取物理节点相在一致性hash环中当前位置的前一个物理节点
	 * @param physicalNode -- 当前位置的物理节点编号
	 * @return -- 前一个物理节点编号
	 * @memo 这里的相对位置都是在一致性hash环上的位置. 
	 */
	public int prevPhysicalNode(int physicalNode) {
		
		checkIndexRange(physicalNode);
		
		int logicalNode = getLogicNode(physicalNode);		
		SortedMap<Integer, Integer> heads = physicalNodeRing.headMap(Integer.valueOf(logicalNode));
		if (heads.isEmpty()) {
			Integer last = physicalNodeRing.get(physicalNodeRing.lastKey());			
			return last.intValue();
		} else {
			Integer last = heads.get(heads.lastKey());
			return last.intValue();
		}
	}
	
	/**
	 * 获取物理节点在一致性hash环中当前位置的下一个物理节点
	 * @param physicalNode -- 当前位置的物理节点编号
	 * @return -- 下一个物理节点编号
	 */
	public int nextPhysicalNode(int physicalNode) {
		
		checkIndexRange(physicalNode);
		
		int logicalNode = getLogicNode(physicalNode);
		SortedMap<Integer, Integer> tails = physicalNodeRing.tailMap(Integer.valueOf(logicalNode + 1)); // 加1是为了将自己排除在外
		if (tails.isEmpty()) {
			Integer first = physicalNodeRing.get(physicalNodeRing.firstKey());
			return first.intValue();
		} else {
			Integer first = tails.get(tails.firstKey());
			return first.intValue();
		}
	}
	
	/**
	 * 一致性hash环中物理节点的数量
	 * @return -- 物理节点数量
	 */
	public int size() {
		
		return physicalNodeRing.size();
	}
}
